/** sort merge join algorithm **/

package qp.operators;

import qp.utils.*;

import java.io.*;
import java.util.*;
import java.lang.*;

import  qp.optimizer.*;
import java_cup.internal_error;


public class SortMergeJoin extends Join{


	int batchsize;  //Number of tuples per out batch

    /** The following fields are useful during execution of
     ** the NestedJoin operation
     **/
    int leftindex;     // Index of the join attribute in left table
    int rightindex;    // Index of the join attribute in right table

    String lfname;	//The file name of where the left table is materialized after sorting
    String rfname;    // The file name where the right table is materialize after sorting

    String rftempname; // for intermedium result of right table
    static int filenum=0;   // To get unique filenum for this operation

    Batch outbatch;   // Output buffer
    Batch leftbatch;  // Buffer for left input stream
    Batch rightbatch;  // Buffer for right input stream
    ObjectInputStream in; // File pointer to the right hand materialized file

    int lcurs;    // Cursor for left side buffer
    int rcurs; // Cursor for right side buffer
    int oldrcurs;//Cursor for the first tuple in right side block which many tuples with the same value
    int rightTempCounter;// how many block that stored in the rightTemp file
    int leftInput;
    int rightInput;
    boolean eosl;  // Whether end of stream (left table) is reached
    boolean eosr;  // End of stream (right table)
    boolean eosrTemp;
    boolean readRightTemp;//the state that read block from temp file of right table
    boolean writeRightTemp;//the state that write block to the temp file of right table
    boolean startReadTemp;//the state that read the first block from the temp file, we need to set the rcurs to oldrcurs
    //boolean felr;	//Whether the first equal last in the block of right table,if yes, the right table need to be stored in disk
    
    Tuple preTuple;// the previous tuple in left table
    
    int leftBuff, rightBuff;//buffer number used by left and right table
    Block leftBlock, rightBlock;//block used
    
    ObjectInputStream leftois, rightois;//used for reading sorted file, batch object 
    ObjectInputStream rightTempois;
    ObjectOutputStream rightTempoos;
    
    public SortMergeJoin(Join jn){
	super(jn.getLeft(),jn.getRight(),jn.getCondition(),jn.getOpType());
	schema = jn.getSchema();
	jointype = jn.getJoinType();
	numBuff = BufferManager.getBuffersPerJoin();
	leftBuff=1;//usually the buffer only need 1
	rightBuff=BufferManager.getBuffersPerJoin()-2;// debug numBuff-2;// the rest buffer used for right table
    }


    /** During open finds the index of the join attributes
     **  Materializes the right hand side into a file
     **  Opens the connections
     **/

    public String sortInput(Operator op,Attribute attr) {
    	int i=0;
    	if(!op.open()) {
    		System.out.println("Error when sortInput() open ()");
    		return null;
    	}
    	else {
    		filenum++;
    		String unSortedFile="unSortedFile-"+String.valueOf(filenum);
    		try {
    			FileOutputStream fos=new FileOutputStream(unSortedFile);
    			ObjectOutputStream oos=new ObjectOutputStream(fos);
    			
    			int tupleSize=op.getSchema().getTupleSize();
    			int batchSize=Batch.getPageSize()/tupleSize;
    			Batch tempOutputBatch=new Batch(batchSize);
    			
    			/**get all contains of the table, and store them in a file*/
    			while((tempOutputBatch=op.next())!=null) {
    				int tempSize=tempOutputBatch.size();
    				i=i+tempSize;//tempOutputBatch.size();
 //   				System.out.println("AAAAAAAAAAAAAAAA-"+i+"-AAAAAAAAAAAA");
    				oos.writeObject(tempOutputBatch);
    				tempOutputBatch=new Batch(batchSize);
    			}
    			fos.close();
    			oos.close();
    		}
    		catch (Exception e) {
    			// TODO: handle exception
    			e.printStackTrace();
    		}
    		
    		String sortedFile = null;
    		Vector<Attribute> attrVector=new Vector<Attribute>();//convert attr to Vector
    		attrVector.add(attr);
    		ExternalSort es=new ExternalSort(unSortedFile,attrVector,op.getSchema());
    		try {
    			sortedFile=es.sort();
			
    		}
    		catch (ArrayIndexOutOfBoundsException e) {
				// TODO: handle exception
    			e.printStackTrace();
				System.out.println("ArrayIndexOutOfBoundsException e");
			}
    		catch (Exception e) {
    			// TODO: handle exception
    			System.out.println("Error when external sort sortInput()");
    			e.printStackTrace();
    		}	
    		return sortedFile;
    	}
    }


    public boolean open(){

		/** select number of tuples per batch **/
		int tuplesize=schema.getTupleSize();
		batchsize=Batch.getPageSize()/tuplesize;

		/*get two attribute from the join::condition*/
		Attribute leftattr = con.getLhs();
		Attribute rightattr =(Attribute) con.getRhs();
		
		/*two attribute -> left/right -> schema  -> index of attritube*/
		leftindex = left.getSchema().indexOf(leftattr);
		rightindex = right.getSchema().indexOf(rightattr);
		
		//Batch rightpage;
		/** initialize the cursors of input buffers **/
	
		lcurs = -1; rcurs =-1; 
		oldrcurs=-1;//old right cursor -1:no record old cursor
		eosl=false;
		/** because right stream is to be repetitively scanned
		 ** if it reached end, we have to start new scan
		 **/
		eosr=false;
		
		eosrTemp=true;
		readRightTemp=false;
		writeRightTemp=false;
		startReadTemp=false;
		rightTempCounter=0;
		leftInput=0;
		rightInput=0;
		/** Right hand side table is to be materialized
		 ** for the Nested join to perform
		 **/
	
		//felr=false;
		
		rightTempois=null;
		rightTempoos=null;
		
		leftBlock=new Block(leftBuff);
		rightBlock=new Block(rightBuff);
		
		if(!left.open() || !right.open()) {
			return false;
		}
		else {		
			lfname=this.sortInput(left, leftattr);			
			rfname=this.sortInput(right, rightattr);
//			System.out.println("QQQQQQQQQQQQQQQQQQQQQQQQQq"+lfname);
//			System.out.println("QQQQQQQQQQQQQQQQQQQQQQQQQq"+rfname);
		}
		if(lfname==null || rfname==null) {
			return false;
		}
		else {
			try {
				FileInputStream leftfis=new FileInputStream(lfname);
				FileInputStream rightfis=new FileInputStream(rfname);
				leftois=new ObjectInputStream(leftfis);
				rightois=new ObjectInputStream(rightfis);
			}
			catch (Exception e) {
				// TODO: handle exception
				System.out.println("error when create FileInputStream in open()");
				e.printStackTrace();
			}
			return true;
		}
    }
	
    /*
    //Test the External Sort
    public Batch next() {
    	outbatch=new Batch(batchsize);
    	if((eosl && lcurs==-1) || (eosr && rcurs==-1)) {
    		close();
    		return null;
    	}
    	if(lcurs==-1) {
        	for(int i=0;i<leftBuff;i++) {
        		try {
    				if (eosl) break;
        			//if(leftois.readObject()!=null) {
        				leftBlock.add(i,(Batch)leftois.readObject());
        			//}
        		}
    			catch(java.io.EOFException f)
    			{
    				eosl=true;				    		
    			}
        		catch(Exception e){
        			e.printStackTrace();	    				
        		}	
    		}
    		if(leftBlock.existTupleInBlock>0) {
    			lcurs++;
    			System.out.println("leftBlock.existTupleInBlock="+leftBlock.existTupleInBlock+"-and lcurs="+lcurs);
    		}
    		
    		for (int i = 0; i < rightBuff; i++) {
    			try {
    				if (eosr)
    					break;
    				//	if(rightois.readObject()!=null) {
    				rightBlock.add(i, (Batch) rightois.readObject());
    				//	}	
    				
    			} catch (java.io.EOFException f) {
    				eosr = true;
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    		if(rightBlock.existTupleInBlock>0) {
    			rcurs++;
    			System.out.println("rightBlock.existTupleInBlock="+rightBlock.existTupleInBlock+"-and rcurs="+rcurs);
    		}
    	}

		for(int i=lcurs;i<leftBlock.existTupleInBlock;i++) {
			Tuple leftTuple=leftBlock.elementAt(i);
			Tuple rightTuple=rightBlock.elementAt(0);
			Tuple outpuTuple=leftTuple.joinWith(rightTuple);
			outbatch.add(outpuTuple);
			if(lcurs==leftBlock.existTupleInBlock-1) {
				lcurs=-1;
			}
			else lcurs++;
			if(outbatch.isFull()) {
				return outbatch;
			}
		}
		return outbatch;
    
    }
    */
    /** from input buffers selects the tuples satisfying join condition
     ** And returns a page of output tuples
     **/

    
    //do NOT consider duplicate
    public Batch next() {
    	if((eosl && lcurs==-1) || (eosr && rcurs==-1)) {
    		close();
    		return null;
    	}
    	outbatch=new Batch(batchsize);
    	while(!outbatch.isFull()) {
    		if((eosl && lcurs==-1) || (eosr && rcurs==-1)) {
    			break;
    		}
    		
    		if(lcurs==-1) {
    			leftBlock.clear();
    			for(int i=0;i<leftBuff;i++) {
    	    		try {
    					if (eosl) break;
    	    			//if(leftois.readObject()!=null) {
    	    				leftBlock.add(i,(Batch)leftois.readObject());
    	    			//}
    	    		}
    				catch(java.io.EOFException f)
    				{
    					eosl=true;				    		
    				}
    	    		catch(Exception e){
    	    			e.printStackTrace();	    				
    	    		}	
    			}
    			if(leftBlock.existTupleInBlock>0) {
    				lcurs++;
 //   				System.out.println("leftBlock.existTupleInBlock="+leftBlock.existTupleInBlock+"-and lcurs="+lcurs);
    			}
    		}
    		
    		if(rcurs==-1) {
    			rightBlock.clear();
 				for (int i = 0; i < rightBuff; i++) {
					try {
						if (eosr)
							break;
						//	if(rightois.readObject()!=null) {
						rightBlock.add(i, (Batch) rightois.readObject());
						//	}	
						
					} catch (java.io.EOFException f) {
						eosr = true;
					} catch (Exception e) {
						e.printStackTrace();
					}
    			}
 				if(rightBlock.existTupleInBlock>0) {
 					rcurs++;
//    				System.out.println("rightBlock.existTupleInBlock="+rightBlock.existTupleInBlock+"-and rcurs="+rcurs);
 				}
    		}
    		
    		while(lcurs!=-1 && rcurs!=-1) {		
    			Tuple leftTuple=leftBlock.elementAt(lcurs);
    			Tuple rightTuple=rightBlock.elementAt(rcurs);
  //  			System.out.println(leftTuple.dataAt(leftindex).toString());
  //  			System.out.println(rightTuple.dataAt(rightindex).toString());
 //   			System.out.println("~~~~~~~~lcurs="+lcurs);
   // 			System.out.println("~~~~~~~~rcurs="+rcurs);

    			//result == 0
				if (Tuple.compareTuples(leftTuple, rightTuple, leftindex,rightindex)==0) {
					Tuple outpuTuple=leftTuple.joinWith(rightTuple);
					outbatch.add(outpuTuple);	
//					System.out.println("~~~~~~Matching result~~~~~~~~");
					if(rcurs==(rightBlock.existTupleInBlock-1)) {
						rcurs=-1;
					}
					else {
						rcurs++;
					}
				}
				//result <0
				else if(Tuple.compareTuples(leftTuple, rightTuple, leftindex,rightindex)<0){
					if(lcurs==(leftBlock.existTupleInBlock-1)) {
						lcurs=-1;
					}
					else {
						lcurs++;				
					}
				}
				//result >0
				else{
					if(rcurs==(rightBlock.existTupleInBlock-1)) {
						rcurs=-1;
					}
					else {
						rcurs++;
					}				
				}	

    			
    			
	    		if(outbatch.isFull()) {
	    			return outbatch;
	    		}    			
    		}
    	}
    	if(outbatch.size()>0) {
    		return outbatch;	
    	}
    	else {
			return null;
		}
    	
    }
    
    
    /*
     //CONSIDER DUPLICATE
    public Batch next(){
	//System.out.print("NestedJoin:--------------------------in next----------------");
	//Debug.PPrint(con);
	//System.out.println();
	if((eosl && lcurs==-1) || (eosr && rcurs==-1)){
		close();
	    return null;
	}
	outbatch = new Batch(batchsize);


	while(!outbatch.isFull()){
		//if the outbatch is NOT full but either left file or right file end, just output the unfull batch
		if((eosl && lcurs==-1) || (eosr && rcurs==-1)) {
			break;
		}
		//fill in the left block for left table
		if(lcurs==-1) {
			leftBlock.clear();
			for(int i=0;i<leftBuff;i++) {
	    		try {
					if (eosl) break;
	    			//if(leftois.readObject()!=null) {
	    				leftBlock.add(i,(Batch)leftois.readObject());
	    				leftInput++;
	    			//}
	    			//else {
	    				//eosl=true;
	    				break;
	    			//}
	    		}
				catch(java.io.EOFException f)
				{
					eosl=true;				    		
				}
	    		catch(Exception e){
	    			e.printStackTrace();	    				
	    		}	
			}
			//for debug
			System.out.println(leftBlock.existTupleInBlock);
			
			if(leftBlock.existTupleInBlock>0) {
				lcurs++;
			}
		}
		
		//fill in the right block for right table
		if(rcurs==-1) {
			rightBlock.clear();
			if(oldrcurs==-1) {
				if(readRightTemp) {
					try {
						rightBlock=(Block)rightTempois.readObject();
						System.out.println("read righblock size"+rightBlock.existTupleInBlock);
						rightTempCounter--;
						if(rightTempCounter==0) {
							readRightTemp=false;
							eosrTemp=true;
						}
					}
					catch (EOFException e) {
						// TODO: handle exception
						rightTempois=null;
						eosrTemp=true;
						readRightTemp=false;
					}
					catch (Exception e) {
						// TODO: handle exception
						e.printStackTrace();
					}
				}
				else {
					for (int i = 0; i < rightBuff; i++) {
						try {
							if (eosr)
								break;
							//	if(rightois.readObject()!=null) {
							rightBlock.add(i, (Batch) rightois.readObject());
							rightInput++;
							//	}	
								//else {
								//	eosr=true;
								//	break;
								//}
						} catch (java.io.EOFException f) {
							eosr = true;
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			}
			
			if(oldrcurs!=-1) {
				if(readRightTemp) {
					try {
						rightBlock=(Block)rightTempois.readObject();
						rightTempCounter--;
					}
					catch (EOFException e) {
						// TODO: handle exception
						rightTempois=null;
						eosrTemp=true;
						readRightTemp=false;
					}
					catch (Exception e) {
						// TODO: handle exception
						e.printStackTrace();
					}
					
					if(startReadTemp) {
						rcurs=oldrcurs;
						startReadTemp=false;
					}				
					
				}
				

				if(!readRightTemp) {
					for (int i = 0; i < rightBuff; i++) {
						try {
							if (eosr)
								break;
							//	if(rightois.readObject()!=null) {
							rightBlock.add(i, (Batch) rightois.readObject());
							rightInput++;
							//	}	
								//else {
								//	eosr=true;
								//	break;
								//}
						} catch (java.io.EOFException f) {
							eosr = true;
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				}
			}
			
			if(rightBlock.existTupleInBlock>0 && rcurs==-1) {
				rcurs++;
			}
		}
		
		while(lcurs!=-1 && rcurs!=-1) {		
			Tuple leftTuple=leftBlock.elementAt(lcurs);
			Tuple rightTuple=rightBlock.elementAt(rcurs);
			System.out.println("++++++++++++++++++New Comparison++++++++++++++++++++++");
			System.out.println(leftTuple.dataAt(leftindex).toString());
			System.out.println(rightTuple.dataAt(rightindex).toString());
			System.out.println("leftblock size="+leftBlock.existTupleInBlock);
			System.out.println("rightblock size="+rightBlock.existTupleInBlock);
			System.out.println("lcurs="+lcurs);
			System.out.println("rcurs="+rcurs);
			System.out.println("===left input buffer NO::"+leftInput);
			System.out.println("===right input buffer NO::"+rightInput);
			try {
				if (Tuple.compareTuples(leftTuple, rightTuple, leftindex,rightindex)==0) {
					Tuple outpuTuple=leftTuple.joinWith(rightTuple);
					outbatch.add(outpuTuple);
					if(oldrcurs==-1) {
						oldrcurs=rcurs;
					}
					
					if(rcurs==rightBlock.existTupleInBlock-1) {
						rcurs=-1;
						if(writeRightTemp) {
							rightTempoos.writeObject(rightBlock);
							rightTempCounter++;
						}
						if(!readRightTemp && !writeRightTemp) {
							filenum++;
							rftempname="RightTempFile-"+String.valueOf(filenum)+".temp";
							FileOutputStream fos=new FileOutputStream(rftempname);
							rightTempoos=new ObjectOutputStream(fos);
							rightTempoos.writeObject(rightBlock);
							System.out.println("write block size="+rightBlock.existTupleInBlock);
							rightTempCounter++;
							writeRightTemp=true;
						}
					}
					else {
						rcurs++;
					}
				}
				else if(Tuple.compareTuples(leftTuple, rightTuple, leftindex,rightindex)<0){
					if(lcurs==leftBlock.existTupleInBlock-1) {
						lcurs=-1;
					}
					else {
						lcurs++;				
					}
					if(writeRightTemp) {
						rightTempoos.writeObject(rightBlock);
						System.out.println("write block size="+rightBlock.existTupleInBlock);
						rightTempCounter++;
						rightTempoos.close();
						rightTempoos=null;
						writeRightTemp=false;
						readRightTemp=true;
						FileInputStream fis=new FileInputStream(rftempname);
						rightTempois=new ObjectInputStream(fis); 
						eosrTemp=false;
						startReadTemp=true;
						rcurs=-1;
					}
					if(readRightTemp) {
						rcurs=-1;
					}
				}
				else{
					if(rcurs==rightBlock.existTupleInBlock-1) {
						rcurs=-1;
					}
					else {
						rcurs++;
					}
					if(oldrcurs!=-1) {
						oldrcurs=-1;
					}
					
				}

			}
			catch (Exception e) {
				// TODO: handle exception
				System.out.println("error when do comparison left right tuple");
				e.printStackTrace();
			}
			System.out.println("---------After one comparison--------");
			System.out.println("writeRightTemp="+writeRightTemp);
			System.out.println("readRightTemp="+readRightTemp);
			System.out.println("eosrTemp="+eosrTemp);
			System.out.println("lcurs="+lcurs);
			System.out.println("rcurs="+rcurs);
			System.out.println("oldrcurs="+oldrcurs);
			// return the full batch
			if(outbatch.isFull()) {
				return outbatch;
			}
		}
		

		
	}
	//return the unfilled batch
	return outbatch;
    }	
	*/

    /** Close the operator */
    public boolean close(){
   File l = new File(lfname);
    File f = new File(rfname);
	l.delete();
    f.delete();
	return true;

    }


}